feat: build gate + inert wiring for contrib Delta scans [Delta contrib split, part 2]#4952
feat: build gate + inert wiring for contrib Delta scans [Delta contrib split, part 2]#4952schenksj wants to merge 1 commit into
Conversation
…b split, part 2] Part 2 of the Delta Lake contrib PR breakup (tracking: apache#4366). Establishes the `contrib-delta` build gate and the inert wiring that lets a gated build compile end to end, while the DEFAULT build stays byte-for-byte unchanged (zero Delta surface). No real Delta read logic yet -- that lands in later parts; here a Delta read that reaches native returns a clean "not implemented" error and falls back to vanilla Spark. Build gate: - Maven `contrib-delta` profile (spark/pom.xml) with per-Spark `delta.version` (3.5->3.3.2, 4.0->4.0.0, 4.1->4.1.0) and an add-source of contrib/delta/src. Default `delta.version` floor in pom.xml. The default spark.version stays 4.1.2 (the delta-spark 4.1.1 pin is a separate, deferred decision). - Cargo `contrib-delta` feature on core (optional path dep on comet-contrib-delta); `native/Cargo.toml` excludes ../contrib from the workspace. - `dev/verify-contrib-delta-gate.sh` proves default cargo/mvn/dylib carry zero Delta surface and the gated build pulls the right deps; wired into a minimal `delta_build_gate.yml` CI job (the full suite/regression workflows land later). Hardened the script against a `set -o pipefail` + `grep -q` SIGPIPE misfire (early grep exit -> echo SIGPIPE -> false guard failure) via here-strings. Inert wiring: - Proto: `Delta*` messages + `delta_scan = 118` (117 is BroadcastNestedLoopJoin). - Native dispatch: `OpStruct::DeltaScan` arm with a not-compiled-in error on default builds and a feature-gated `delta_scan` shim that calls the contrib; exhaustive-match arms in operator_registry/jni_api; `convert_spark_types_to_ arrow_schema` promoted to pub(crate). - Stub contrib crate (contrib/delta/native): `plan_delta_scan` returns `DataFusionError::NotImplemented` -- just enough to satisfy the core shim's contract so `--features contrib-delta` links. - JVM bridge `DeltaIntegration` (reflective, all lookups return None until the contrib classes exist), the CometExecRule Delta-marker hook (CDF hook deferred to a later part), the CometScanRule Delta delegation + metadata-col reorder, and the leaf `DeltaConf`. Verification: default + gated native build, clippy both feature states, gate script, gated + default JVM compile, spotless/scalastyle, cargo fmt -- all green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
@parthchandra @andygrove this is part 2 of the Delta contrib split — the build gate + inert wiring — following on from part 1 (#4700, now merged). Thank you both for the reviews on part 1! 🙏 Sorry it took a while to get this next phase out there — I was heads-down getting https://github.com/capitalone/vulnhunter to market. Back on the Delta series now. This part is deliberately inert/gated (zero Delta surface on default builds), so it should be a fairly self-contained review of the wire format, the build machinery, and the gate-enforcement script. Would appreciate your eyes when you have a chance. |
| * and the lookups resolve, dispatching the call into the contrib helpers. | ||
| * | ||
| * Keeping this bridge as one small file in core lets the Delta detection block in `CometScanRule` | ||
| * and the serde dispatch in `CometExecRule` stay ~10 lines each -- exactly the shape Parth's |
There was a problem hiding this comment.
The change to CometExecRule is small but this itself is pure Delta code in core. We really need to follow the same pattern we established of discovering using the ServiceLoader mechanism. PlanDataInjector (in operators.scala:72-127, merged in Part 1 / #4700) already uses ServiceLoader discovery. PR #4633 (Lance) also uses the same pattern with CometScanContrib
I would recommend we replace this with a generic CometScanContrib trait in a new file spark/src/main/scala/org/apache/comet/rules/CometScanContrib.scala (in core). This trait can cover both V1 and V2 scans:
trait CometScanContrib {
/** V1 scan hook. Return Some(plan) to claim, None to pass. */
def tryTransformV1(
plan: SparkPlan,
session: SparkSession,
scanExec: FileSourceScanExec,
relation: HadoopFsRelation): Option[SparkPlan] = None
/** V2 scan hook. Return Some(plan) to claim, None to pass. */
def tryTransformV2(scanExec: BatchScanExec): Option[SparkPlan] = None
}
object CometScanContrib extends Logging {
private lazy val contribs: Seq[CometScanContrib] = {
// Built-in contribs (Parquet, Iceberg) can be registered here.
// Contrib-gated ones (Delta, Lance) discovered via ServiceLoader.
val discovered = try {
ServiceLoader.load(classOf[CometScanContrib], getClass.getClassLoader).asScala.toSeq
} catch {
case NonFatal(e) =>
logWarning("Failed to load contrib CometScanContrib services", e)
Seq.empty
}
discovered
}
def tryTransformV1(...): Option[SparkPlan] = {
contribs.view.flatMap(_.tryTransformV1(...)).headOption
}
def tryTransformV2(...): Option[SparkPlan] = {
contribs.view.flatMap(_.tryTransformV2(...)).headOption
}
}
This is format-agnostic code.
The Delta contrib then has:
contrib/delta/src/.../DeltaScanRuleContrib.scalaimplementingCometScanContribcontrib/delta/resources/META-INF/services/org.apache.comet.rules.CometScanContribnaming it
This also subsumes PR #4633's CometScanContrib — the Lance PR would implement tryTransformV2 on the same trait rather than defining a separate one.
| // Delta Lake scan. Wire format used by `contrib/delta/`. Only decoded when | ||
| // core is built with `--features contrib-delta`; in default builds the | ||
| // dispatcher arm is `#[cfg]`-stubbed out so the contrib has zero runtime cost. | ||
| DeltaScan delta_scan = 118; |
There was a problem hiding this comment.
Currently both this PR and PR #4633 claim field number 118 (delta_scan = 118 vs lance_scan = 118). More fundamentally, adding a new oneof variant for every contrib scan type means the core proto file must change every time a new format is added — violating the goal that core doesn't know about contrib implementations.
Spark Connect solves this at relations.proto:109 with google.protobuf.Any extension = 998;. We should do the same.
We can replace this change with a permanent extension point:
// One-time addition to the oneof, never changes again:
google.protobuf.Any contrib_scan = 200;
Then DeltaScan and LanceScan message definitions still live in operator.proto (or preferably in separate proto files under contrib/) — they're just packed into Any on the JVM side and unpacked by type_url on the Rust side.
There was a problem hiding this comment.
If this gets tricky to implement we can consider assigning field ids for contrib scans, but I feel that the approach outlined here is feasible.
| case scan if !CometConf.COMET_NATIVE_SCAN_ENABLED.get(conf) => | ||
| withFallbackReason(scan, "Comet Scan is not enabled") | ||
|
|
||
| // V1 scans go through `transformV1Scan` which itself first delegates to any |
There was a problem hiding this comment.
With a generic CometScanContrib.tryTransformV1(), this reordering is unnecessary. The outer transformScan match keeps its current order (metadata-colum guard before FileSourceScanExec). Each contrib's tryTransformV1 can decide internally whether it can handle metadata columns.
| // vanilla scan path. When the Delta classes are on the classpath, the contrib | ||
| // either claims the scan (returning a CometScanExec marker) or declines via | ||
| // its own `withFallbackReason` fallback message. | ||
| DeltaIntegration.transformV1IfDelta(plan, session, scanExec, r) match { |
There was a problem hiding this comment.
This will change to CometScanContrib.tryTransformV1(plan, session, scanExec, r). We can also remove the re-applied metadataCols check — it's now redundant because the outer guard still runs first and the Delta implementation decides whether to handle metadata columns or not.
| // activated. The marker wraps the original, link-bearing scan, so the produced exec's | ||
| // originalPlan keeps its logicalLink with no workaround. If conversion declines, the marker | ||
| // itself falls back to the vanilla Spark Delta scan, so leaving it in the plan is safe. | ||
| case scan if DeltaIntegration.isDeltaScanMarker(scan) => |
There was a problem hiding this comment.
We can clean this up too - Define a marker trait and Delta can implement it
/** Marker trait for contrib scan nodes that carry their own serde handler. */
trait CometContribScanMarker { this: SparkPlan =>
def scanHandler: CometOperatorSerde[_ <: SparkPlan]
}
then the match becomes
case marker: CometContribScanMarker =>
convertToComet(marker, marker.scanHandler).getOrElse(marker)
2 participants
Part 2 of the Delta Lake contrib PR breakup. Part 1 (#4700, the core SPI for contrib leaf scans) is merged. This part establishes the
contrib-deltabuild gate and the inert wiring that lets a gated build compile and link end to end — while the default build stays byte-for-byte unchanged (zero Delta surface). It ships no real Delta read logic: a Delta read that reaches native returns a clean "not implemented" error and falls back to vanilla Spark. The full sequence and dependency graph live in the tracking umbrella, #4366.The whole point of this part is that everything it adds is gated or inert, so it is safe to land on
mainwell ahead of the read path, and reviewers can review the wire format, the build machinery, and the gate enforcement once, in isolation, before any Delta code shows up.Changes
Build gate
contrib-deltaprofile (spark/pom.xml) with a per-Sparkdelta.version(3.5 → 3.3.2, 4.0 → 4.0.0, 4.1 → 4.1.0) and anadd-sourceofcontrib/delta/src. Defaultdelta.versionfloor inpom.xml. The defaultspark.versionstays 4.1.2 — the delta-spark 4.1.1 pin is a separate, deferred decision (raised for later parts).contrib-deltafeature on core (optional path dep oncomet-contrib-delta);native/Cargo.tomlexcludes../contribfrom the workspace so non-Delta committers never build it.dev/verify-contrib-delta-gate.shproves the default cargo tree, Maven dependency set, compiled classes, andlibcometall carry zero Delta surface, and that the gated build pulls the right deps (delta-spark per Spark profile,comet-contrib-deltain the cargo tree). Wired into a minimaldelta_build_gate.ymlCI job. The full test-suite and regression workflows land in later parts.Inert wiring
Delta*messages anddelta_scan = 118(117 isBroadcastNestedLoopJoin). One-time wire-format review here, early.OpStruct::DeltaScanarm that returns a not-compiled-in error on default builds, and a feature-gateddelta_scanshim that calls the contrib; exhaustive-match arms inoperator_registry/jni_api;convert_spark_types_to_arrow_schemapromoted topub(crate).contrib/delta/native):plan_delta_scanreturnsDataFusionError::NotImplemented— just enough to satisfy the core shim's contract so--features contrib-deltalinks. This makes the exact core↔contrib contract visible in a small PR.DeltaIntegration(reflective; every lookup returnsNoneuntil the contrib classes exist), theCometExecRuleDelta-marker hook (the CDF hook is deferred to a later part), theCometScanRuleDelta delegation + metadata-column reorder, and the leafDeltaConf.What this part deliberately does NOT do yet
plan_delta_scanis a stub. Log replay, predicate pushdown, deletion vectors, and the kernel read land in parts 3a/3b.DeltaIntegration's lookups resolve toNone, so the marker always falls back. The claim/decline layer and native exec land in parts 4a/4b.CometExecRuleCDF hook andDeltaIntegration's CDF members are held back to part 5.Why it is safe on default builds
Without
-Pcontrib-delta: the cargo feature is off (nocomet-contrib-delta, nodelta_kernelin the tree), the Maven profile is inactive (noio.delta:*, nocontrib/deltasources compiled), and the nativeDeltaScanarm is#[cfg]-stubbed to an error that is never reached because nothing emits the proto message.DeltaIntegrationis the only always-present class and every one of its reflective lookups returnsNone. The gate script asserts all of this mechanically: defaultlibcomethas 0 Delta symbols and is the same size as before.The
CometScanRulechange reorders the metadata-column guard so V1 scans reachtransformV1Scan(which delegates to any V1 contrib) before the generic metadata-column rejection — but for a non-contrib V1 scan the guard is re-applied insidetransformV1Scan, so vanilla behavior is unchanged.Verification
Run against this branch rebased on current
main:--features contrib-delta): green.cargo clippyboth feature states: clean.dev/verify-contrib-delta-gate.sh: all checks pass (0 Delta symbols in the default dylib).-Pcontrib-delta) and default JVM compile (spark-3.4 / Scala 2.12): both compile.Roadmap
Parts still to come: Rust driver-side planning (3a), Rust executor-side read path (3b), Scala claim/decline (4a), Scala execution — end-to-end native reads (4b), Change Data Feed (5), test battery + regression harness (6), and docs (7). Each is gated behind
-Pcontrib-delta, so every intermediate state onmainis safe for default builds. Tracking umbrella: #4366; part 1: #4700.🤖 AI disclosure: this PR was prepared with assistance from Claude Code (Claude Opus 4.8), under the submitter's review and direction.